package com.homedo.rxj;

import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.Observer;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.Action;
import io.reactivex.functions.Consumer;
import io.reactivex.schedulers.Schedulers;

import java.util.concurrent.TimeUnit;

public class RxjTest3 {
    static ObservableEmitter<Integer> curEmitter = null;

    public static void main(String[] args) throws InterruptedException {

        Observable observable = createObserver().doOnComplete(new Action() {
            @Override
            public void run() throws Exception {
                System.out.println("Completed");
            }
        }).doOnError(new Consumer<Throwable>() {
            @Override
            public void accept(Throwable throwable) throws Exception {
                System.out.println("Error:" + throwable.getMessage());
            }
        });

        observable.window(Observable.interval(10, TimeUnit.SECONDS, Schedulers.newThread())).subscribe(new Observer() {
            @Override
            public void onSubscribe(Disposable d) {

            }

            @Override
            public void onNext(Object o) {
                System.out.println(o);
            }

            @Override
            public void onError(Throwable e) {

            }

            @Override
            public void onComplete() {

            }
        });

        Disposable disposable = observable.subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                System.out.println(integer);
            }
        });


        for (int i = 0; i < 100; i++) {
            if (!curEmitter.isDisposed()) {
                curEmitter.onNext(i);
                if (i == 100 - 1) {
                    curEmitter.onComplete();
                }
                Thread.sleep(1000L);
            }
        }
    }


    private static Observable<Integer> createObserver() {
        return Observable.create(
                new ObservableOnSubscribe<Integer>() {
                    @Override
                    public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                        curEmitter = emitter;
                    }
                }
        );
    }
}
